package net.bwie.realtime.vehicle.dwd.utils;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;

/**
 * @author ASUS
 */
public class MonitorUtil {
    // JSON解析器（工具类内部使用）
    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
    // 限速信息缓存（工具类内部使用）
    private static final Map<String, Integer> MONITOR_INFO_CACHE = new HashMap<>();

    /**
     * 预加载区域信息到缓存
     */
    public static void preloadAreaInfo() {
        // 数据库连接信息（请修改为你的实际配置）
        String url = "jdbc:postgresql://node103:5432/smart_traffic";
        String user = "postgres";
        String password = "123456";

        Connection conn = null;
        Statement stmt = null;
        ResultSet rs = null;

        try {
            // 加载PostgreSQL驱动
            Class.forName("org.postgresql.Driver");
            conn = DriverManager.getConnection(url, user, password);
            stmt = conn.createStatement();
            rs = stmt.executeQuery("SELECT monitor_id, limit_speed FROM public.t_monitor_info");

            // 存入缓存
            while (rs.next()) {
                String monitorId = rs.getString("monitor_id");
                Integer speedLimit = rs.getInt("limit_speed");
                MONITOR_INFO_CACHE.put(monitorId, speedLimit);
            }

            System.out.println("预加载限速信息完成，共加载 " + MONITOR_INFO_CACHE.size() + " 条数据");

        } catch (Exception e) {
            System.err.println("预加载失败: " + e.getMessage());
            e.printStackTrace();
        } finally {
            // 关闭资源
            try { if (rs != null) {
                rs.close();
            }
            } catch (Exception e) { e.printStackTrace(); }
            try { if (stmt != null) {
                stmt.close();
            }
            } catch (Exception e) { e.printStackTrace(); }
            try { if (conn != null) {
                conn.close();
            }
            } catch (Exception e) { e.printStackTrace(); }
        }
    }

    /**
     * 处理JSON流数据，添加区域名称
     */
    public static DataStream<Integer> processJson(DataStream<String> logStream) {
        return logStream.flatMap(new FlatMapFunction<String, Integer>() {
            @Override
            public void flatMap(String s, Collector<Integer> collector) throws Exception {
                try {
                    JsonNode jsonNode = OBJECT_MAPPER.readTree(s);
                    ObjectNode objectNode = (ObjectNode) jsonNode;
                    String monitorId = jsonNode.get("monitor_id").asText();
                    Integer speedLimit = MONITOR_INFO_CACHE.getOrDefault(monitorId, 60);
                    objectNode.put("speed_limit", speedLimit);
                    collector.collect(speedLimit);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

    }

}
